package com.sun.enhancekafka.api;

import com.sun.enhancekafka.config.MyPartitioner;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaApiController {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @GetMapping("/test")
    public void test() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.204.129:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //自定义分区
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());

        //提高kafka的吞吐量的四个参数

        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);  //缓冲区大小
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);  //批次大小
        properties.put(ProducerConfig.LINGER_MS_CONFIG,1);   //linger ms
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");   //压缩

        //ack
        properties.put(ProducerConfig.ACKS_CONFIG,1);
        //重试
        properties.put(ProducerConfig.RETRIES_CONFIG,6);

        //开启幂等性，默认是true,解决数据重复的问题
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);


        KafkaProducer<String,String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            //直接发送
            //kafkaProducer.send(new ProducerRecord<>("myKafka","hello" + i));
            //带回调函数的方法,返回主题，分区信息
//            kafkaProducer.send(new ProducerRecord<>("myKafka", "CallBackFunction" + i), (recordMetadata, e) -> {
//                if(e == null){
//                    System.out.println("主题：" + recordMetadata.topic() + "; 分区：" + recordMetadata.partition());
//                }
//            });

            //同步发送
            // kafkaProducer.send(new ProducerRecord<>("myKafka","TongbuFaSong" + i)).get();

            //发送到1号分区上去
            kafkaProducer.send(new ProducerRecord<>("myKafka",1,"","Partition" + i));
        }

        kafkaProducer.close();
    }

}
